package com.chb.flink.source

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.kafka.common.serialization.StringDeserializer

/**
 * 第一种Kafka Source: 只处理简单数据类型，String
 */
object KafkaSourceByString {
    def main(args: Array[String]): Unit = {
        //初始化Flink的Streaming（流计算）上下文执行环境
        val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
        streamEnv.setParallelism(1)
        //导入隐式转换
        import org.apache.flink.streaming.api.scala._

        // kafka配置
        val props = new Properties()
        props.setProperty("bootstrap.servers", "ShServer:9092")
        props.setProperty("group.id", "chb01")
        props.setProperty("key.deserializer", classOf[StringDeserializer].getName)
        props.setProperty("value.deserializer", classOf[StringDeserializer].getName)
        props.setProperty("auto.offset.reset", "latest")

        //设置kafka为数据源
        val flinkKafkaConsumer = new FlinkKafkaConsumer[String]("test", new SimpleStringSchema(), props)
        val stream = streamEnv.addSource(flinkKafkaConsumer)

        stream.print()
        streamEnv.execute()
    }
}